package generic.activemq;

import generic.Subscriber;

import java.util.HashSet;
import java.util.Set;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnectionFactory;

public class ActiveMqRedundantSubscriber implements MessageListener {

	private Subscriber subscriber;

	private String url1 = "tcp://ec2-54-242-185-60.compute-1.amazonaws.com:61616";
	private String url2 = "tcp://ec2-54-242-185-60.compute-1.amazonaws.com:61618";
//	private String url1 = "tcp://localhost:61616";
//	private String url2 = "tcp://localhost:61618";
	// private String url = "tcp://localhost:61616";
	// private String url =
	// "failover:(tcp://localhost:61616,tcp://localhost:61618)?randomize=false&backup=true&initialReconnectDelay=1&maxReconnectAttempts=1";
	// private String url =
	// "fanout:(static:(tcp://localhost:61616,tcp://localhost:61618))?minAckCount=1&initialReconnectDelay=1&maxReconnectAttempts=1";

	private StringBuilder builder;
	private Set<Integer> messageSet;

	public ActiveMqRedundantSubscriber(Subscriber subscriber, String url1, String url2) {
		super();
		this.subscriber = subscriber;
		this.url1 = url1;
		System.out.println("url1:" + url1);
		this.url2 = url2;
		System.out.println("url2:" + url2);

		builder = new StringBuilder();
		messageSet = new HashSet<Integer>();

		initialize(url1);
		initialize(url2);

		System.out.println("ActiveMqRedundantListener waiting for messages...");
	}

	private void initialize(String brokerUrl) {
		try {
			ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(
					brokerUrl);
			Connection connection;
			connection = factory.createConnection();
			Session session = connection.createSession(false,
					Session.AUTO_ACKNOWLEDGE);
			Topic topic = session.createTopic("topictest.messages");
			MessageConsumer consumer = session.createConsumer(topic);
			consumer.setMessageListener(this);

			connection.start();
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void onMessage(Message jmsMessage) {
		synchronized (this) {
			String message = "";
			try {
				TextMessage textMessage = (TextMessage) jmsMessage;
				message = textMessage.getText();
			} catch (JMSException e) {
				e.printStackTrace();
			}

			int messageId = Integer.valueOf(message.split(";")[0]);
			if (messageSet.add(messageId)) {
				subscriber.receive(message);
			}
		}

	}
}
